package reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.*;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.List;

/**
 * @author： zhangnan
 * @date： 2025/5/22 19:42
 * @description： TODO
 * @modifiedBy：
 * @version: 1.0
 */
public class FluxDemo {

    /**
     * Reactor中数据分为2种
     * 没有或单个(0或1个元素)数据 - Mono[0|1]
     * 多个数据  - Flux[N]
     */
    public void test() throws IOException {
//        Flux.just(1,2,3)
//                .delayElements(Duration.ofSeconds(1))  // 调频操作 - 延迟指定的时间下发数据

        Flux.concat(Flux.just(1, 2), Mono.just(2)) // 合并两个数据流
                .delayElements(Duration.ofSeconds(1)) // 调频操作 - 延迟指定的时间下发数据
                .distinct()
                .log()
                .subscribe();


//        Flux.range(1,8)
//                .log()
//                .filter(i -> i > 3) // 挑出 > 3 的元素
//                .map(i -> "haha-" + i) // 转换
//                .subscribe(System.out::println);
        System.in.read();
    }


    public static void main(String[] args) throws IOException {

        /**
         *
         *  响应式编程：全异步、消息、事件回调，
         *  但是 默认 还是用 当前线程 生成整个 流、发布流、流的中间 操作
         *
         * Schedulers.immediate(); // 默认：无执行上下文，当前线程运行所有操作
         * Schedulers.single(); // 使用固定的一个单线程
         * Schedulers.boundedElastic(); // 有界(不是无限扩充的线程池，线程池中有 10 * CPU 核心个线程，队列默认 100K)，弹性调度
         * Schedulers.newBoundedElastic(); // 自定义线程池参数
         * Schedulers.fromExecutor(); // 入参是 自定义 的线程池
         * Schedulers.parallel(); // 并发池，使用 forkjoin
         *
         * 注意：
         *  只要不指定线程池，默认发布者用的线程就是订阅者的线程
         */
        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);
        Flux<String> flux = Flux.range(1, 10)
                .map(i -> 10 + i)
                .log()
                // 改变发布者所在线程池
//                .publishOn(Schedulers.single()) // 在哪个线程池把这个流的数据和操作执行了；
                .publishOn(s)
                .map(i -> "value" + i)
                .log();
//                .subscribeOn(Schedulers.single()) // 改变订阅者所在线程池

        // 只要不指定线程池，默认发布者用的线程就是订阅者的线程
        new Thread(() -> flux.subscribe(System.out::println)).start();


        // 自定义流中元素处理规则
        Flux.range(1, 10)
                .handle((val, sink) -> {   // 参数一：流中的正要处理的数据；参数二：数据通道引用
                    System.out.println("本次处理流中元素： " + val);
                    sink.next("zhagnsan " + val);  // 可以向下发送数据的通道
                })
                .log()
                .subscribe();

//        new FluxDemo().generateAndCreateAndHandlerTest();

        System.in.read();


    }

    private void generateAndCreateAndHandlerTest() {
        // 编程方式创建序列
        // sink: 接收器、水槽、通道
        // source：数据源，sink： 接受端
        // 同步环境调用 创建 可以使用 generate
        Flux.generate(() -> 0,  // 初始值
                        (val, sink) -> {
                            if (val < 11) {
                                val += 1;
                                sink.next(val); // 把元素传出去 , generate中Sink.next只能调用一次
//                                sink.next(val); // 如果多次调用 会出现异常，如哦需要多次调用，可以使用 Flux.create()
                            } else {
                                sink.complete(); // 完成，完成信号
                            }
                            return val; // 返回新的值
                        })
                .log()
                .subscribe();

        // 异步多线程环境创建 可以调用 create
        Flux.create(sink -> {
                    for (int i = 0; i < 10; i++) {
                        sink.next(i);
                        sink.next("test: " + i);  // 多次调用next() 方法

                        MyListener listener = new MyListener(sink);
                        listener.onLine("张" + i);
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }

                })
                .log()
                .subscribe();


    }


    class MyListener {

        FluxSink<Object> sink;

        public MyListener() {
        }

        public MyListener(FluxSink<Object> sink) {
            this.sink = sink;
        }

        // 用户登录，触发onLine方法
        public void onLine(Object userName) {
            System.out.println("用户 " + userName + " 登录了");
            sink.next(userName);
        }
    }

    private static void limitRateTest() {
        // 按比例限流
        Flux.range(1, 100)
                .log()
                // 75% 预取策略
                // 如果 75% 的元素已经处理了，继续抓取 新的 75% 元素；即 第一次抓取100个数据，以后每次抓取75
                .limitRate(100)
                .subscribe();
    }

    // flux buffer 用法
    private static void bufferTest() {
        // buffer 用法
        // 生成10个元素流
        Flux<List<Integer>> buffer = Flux.range(1, 10)
                .buffer(3); // 缓冲区：缓冲3个元素，消费一次最多可以拿到三个元素，凑满数批量发送给消费者;
        // 10个元素，buffer(3),消费者请求4次，数据消费完成


        buffer.subscribe(new BaseSubscriber<List<Integer>>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
//                        request(N); // 找发布者请求一次数据，一次的数据量是 N * bufferSize，示例中buffer设置为3
                request(1); // 找发布者请求1次数据，示例中buffer设置为3，也就是请求一次数据会返回3条数据
                super.hookOnSubscribe(subscription);
            }

            @Override
            protected void hookOnNext(List<Integer> value) {
                System.out.println(value); // 打印到达数据
                request(1);
//                        super.hookOnNext(value);
            }
        });
    }

    private static void testFluxAndMono() {
        //        Mono: 0|1个元素的流
//        Flux: N个元素的流；  N>1
        //发布者发布数据流：源头
        // 流不消费就没用； 消费：订阅
        Flux<Integer> just = Flux.just(1, 2, 3);// 多元素流

        just.subscribe(e -> System.out.println("e1 = " + e));
        // 一个数据流可以有很多消费者
        //对于每个消费者来说流都是一样的； 类似mq中的：广播模式；
        just.subscribe(e -> System.out.println("e2 = " + e));


        System.out.println("==================");
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1)); // 从0开始每秒产生一个递增的数字

        flux.subscribe(System.out::println); // 订阅数据
    }

    /**
     * doOnxxx：发生这个事件的时候产生一个回调，通知你
     * onxxx：发生这个事件后执行一个动作，可以改变元素、信号
     */
    private static void fluxDoOn() {
        //Mono<T>： 只有零个或一个T
//        Mono.just(1).subscribe(System.out::println);

        //Flux<Integer>： 有很多Integer
        // 空流：链式API中，下面的操作符，操作的是上面的流
        // 事件感知API：当流发生什么事件时，触发一个回调，系统调用提前定义好的钩子函数（Hook【钩子函数】）：doOnxxx
        Flux<Integer> flux = Flux.range(1, 7)
                .delayElements(Duration.ofSeconds(1))
//                .log()
                .doOnComplete(() -> System.out.println("流正常结束..."))
                .doOnCancel(() -> {
                    System.out.println("流已被取消...");
                    int i = 10 / 0;  // 该异常只会触发 doOnError
                })
                .doOnError(throwable -> System.out.println("流异常..." + throwable.getMessage()));

        // 订阅者的几种调用形式
        flux.subscribe(v -> System.out.println("v = " + v));  //
        flux.subscribe(v -> System.out.println("v = " + v),
                throwable -> throwable.printStackTrace());
        flux.subscribe(v -> System.out.println("v = " + v),   // 正常消费流元素
                throwable -> throwable.printStackTrace(),  // 感知异常结束
                () -> System.out.println("流结束了..."));  // 感知正常结束

        // 自定义订阅者
        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅者和发布者绑定好了：" + subscription);
                request(1); // 订阅1条数据,背压
                requestUnbounded(); // 获取无线数据 - 所有数据

//                super.hookOnSubscribe(subscription);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("元素到达： " + value);
                if (value < 5) {
                    request(1);
//                    if (value == 3) {
//                        int i = 10 / 0; // 触发订阅异常,该异常只会触发订阅者的hookOnError事件
//                    }
                } else {
                    cancel(); // 取消订阅
                }
//                super.hookOnNext(value);
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("hookOnComplete");
//                super.hookOnComplete();
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("hookOnError: " + throwable.getMessage());
//                super.hookOnError(throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("hookOnCancel");
//                super.hookOnCancel();
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("hookFinally，信号类型： " + type);
//                super.hookFinally(type);
            }
        });
    }

    /**
     * 响应式编程核心：看懂文档弹珠图；
     * 信号： 正常/异常（取消）
     * SignalType：
     * SUBSCRIBE： 被订阅
     * REQUEST：  请求了N个元素
     * CANCEL： 流被取消
     * ON_SUBSCRIBE：在订阅时候
     * ON_NEXT： 在元素到达
     * ON_ERROR： 在流错误
     * ON_COMPLETE：在流正常完成时
     * AFTER_TERMINATE：中断以后
     * CURRENT_CONTEXT：当前上下文
     * ON_CONTEXT：感知上下文
     * <p>
     * doOnXxx API触发时机
     * 1、doOnNext：每个数据（流的数据）到达的时候触发
     * 2、doOnEach：每个元素（流的数据和信号）到达的时候触发
     * 3、doOnRequest： 消费者请求流元素的时候
     * 4、doOnError：流发生错误
     * 5、doOnSubscribe: 流被订阅的时候
     * 6、doOnTerminate： 发送取消/异常信号中断了流
     * 7、doOnCancle： 流被取消
     * 8、doOnDiscard：流中元素被忽略的时候
     */
    private static void doOnxxxTest() {
        Flux.just(1, 2, 3, 4, 5, 6, 7, 0, 5, 6)
                .doOnNext(item -> System.out.println("元素到达： " + item))  // 当数据到达时触发
                .doOnEach(itemSignal -> {
                    System.out.println("doOnEach: " + itemSignal.get() + ", " + itemSignal.getType());
                }) // 触发时机和doOnNext一致，但doOnEach的元素更加丰富，包含signal
                .map(item -> 10 / item)
                .doOnError(throwable -> System.out.println(throwable.getMessage()))
                .subscribe(); // 不订阅，是不会触发任何事件的
    }

}
